package com.coffee.netty;

import cn.hutool.core.util.StrUtil;
import com.coffee.concurrent.CoffeeThreadFactory;
import com.coffee.netty.config.ConfigManager;
import com.coffee.netty.util.NettyEventLoopUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.ResourceLeakDetector;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @author songkui
 * @since 2021/7/31
 **/
public class CoffeeRpcServer extends AbstractRemotingServer {
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(CoffeeRpcServer.class);

    /** server bootstrap */
    private ServerBootstrap bootstrap;
    /** rpc codec */
    private RpcCodec codec;
    private CommandHandler commandHandler;

    /** boss event loop group, boss group should not be daemon, need shutdown manually*/
    private final EventLoopGroup bossGroup = NettyEventLoopUtil.newEventLoopGroup(
                    1,
                    new CoffeeThreadFactory("Rpc-boss", false));

    /** worker event loop group. Reuse I/O worker threads between rpc servers. */
    private final EventLoopGroup workerGroup  = NettyEventLoopUtil.newEventLoopGroup(
                    Runtime
                            .getRuntime()
                            .availableProcessors() * 2,
                    new CoffeeThreadFactory( "Rpc-worker", true));

    /** channelFuture */
    private ChannelFuture channelFuture;


    @Override
    public void registerRpcCodec(RpcCodec rpcCodec) {
        this.codec = rpcCodec;
    }

    @Override
    public void registerCommandHandler(CommandHandler commandHandler) {
        this.commandHandler = commandHandler;
    }


    @Override
    protected void doInit(){
        this.bootstrap = new ServerBootstrap();
        this.bootstrap.group(bossGroup, workerGroup)
                .channel(NettyEventLoopUtil.getServerSocketChannelClass())
                .option(ChannelOption.SO_BACKLOG, ConfigManager.tcp_so_backlog())
                .option(ChannelOption.SO_REUSEADDR, ConfigManager.tcp_so_reuseaddr())
                .childOption(ChannelOption.TCP_NODELAY, ConfigManager.tcp_nodelay())
                .childOption(ChannelOption.SO_KEEPALIVE, ConfigManager.tcp_so_keepalive());
        initWriteBufferWaterMark();
        // init byte buf allocator
        if (ConfigManager.netty_buffer_pooled()) {
            this.bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        } else {
            this.bootstrap.option(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT)
                    .childOption(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT);
        }

        // enable trigger mode for epoll if need
        NettyEventLoopUtil.enableTriggeredMode(bootstrap);
        final ChannelHandler serverIdleHandler = new HeartbeatServerHandler();
        final RemotingHandler rpcHandler = new RemotingHandler(commandHandler);
        this.bootstrap
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(codec.packDecoder());
                        pipeline.addLast("decoder", codec.decoder());
                        pipeline.addLast("encoder", codec.encoder());
                        pipeline.addLast("idleStateHandler", new IdleStateHandler(ConfigManager.tcp_server_idle(), 0, 0, TimeUnit.SECONDS));
                        pipeline.addLast("serverIdleHandler", serverIdleHandler);
                        pipeline.addLast("handler", rpcHandler);
                    }
                });

        ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.SIMPLE);
    }

    private void initWriteBufferWaterMark() {
        int lowWaterMark = ConfigManager.netty_buffer_low_watermark();
        int highWaterMark = ConfigManager.netty_buffer_high_watermark();
        if (lowWaterMark > highWaterMark) {
            throw new IllegalArgumentException(
                    String.format("[server side] coffee netty high water mark {%s} should not be smaller than low water mark {%s} bytes)",
                                    highWaterMark, lowWaterMark));
        } else {
            logger.warn("[server side] coffee netty low water mark is {} bytes, high water mark is {} bytes",
                    lowWaterMark, highWaterMark);
        }
        this.bootstrap.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
                new WriteBufferWaterMark(lowWaterMark, highWaterMark));
    }


    @Override
    protected boolean doStart() throws InterruptedException {
        if (StrUtil.isNotEmpty(ConfigManager.getInstance().inetHost())){
            this.channelFuture = this.bootstrap.bind(ConfigManager.getInstance().inetHost(), ConfigManager.getInstance().port()).sync();
        }else {
            this.channelFuture = this.bootstrap.bind(ConfigManager.getInstance().port()).sync();
        }

        return this.channelFuture.isSuccess();
    }

    @Override
    protected void doStop() {
        if (null != this.channelFuture) {
            this.channelFuture.channel().close();
        }

        this.bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
        logger.warn("Coffee netty Server stopped!");
    }

}
